package com.company.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import redis.clients.jedis.Jedis;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.jar.JarEntry;

public class CanalSyncDataToRedis {

    static  Jedis jedis = new Jedis("124.70.161.113",6379);

    public static void main(String[] args) throws InvalidProtocolBufferException {

        //创建链接
        CanalConnector connector = CanalConnectors
                .newSingleConnector(new InetSocketAddress("124.70.161.113", 11111),
                        "example",
                        "canal",
                        "canal");
        connector.connect();
        //connector.subscribe(".*\\..*");
//        connector.subscribe("example\\.test,example\\.demo");
        //只订阅example下的test数据库的变化
        connector.subscribe("example\\.test");
        connector.rollback();





        while (true) {

            Message withoutAck = connector.getWithoutAck(10);

            long id = withoutAck.getId();
            int size = withoutAck.getEntries().size();
            if (id == -1 || size == 0) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                process(withoutAck.getEntries());


                connector.ack(id);
            }


        }


    }


    private static void process(List<CanalEntry.Entry> entries) throws InvalidProtocolBufferException {

        jedis.auth("981204");

        for (int i = 0; i < entries.size(); i++) {
            CanalEntry.Entry entry = entries.get(i);

            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
                    entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange = CanalEntry.
                    RowChange.parseFrom(entry.getStoreValue());

            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDatasList) {

                String schemaName = entry.getHeader().getSchemaName();
                String tableName = entry.getHeader().getTableName();
                System.out.printf("数据库: %s , 表:%s \n ", schemaName, tableName);


                if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {
                    List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();

                    System.out.println("删除行before:" + beforeColumnsList);
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    System.out.println("删除行after:" + afterColumnsList);


                } else if (rowChange.getEventType() == CanalEntry.EventType.UPDATE) {
                    List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();

                    System.out.println("更新行before:" + beforeColumnsList);
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    System.out.println("更新行after:" + afterColumnsList);

                    Iterator<CanalEntry.Column> iterator = afterColumnsList.iterator();
                    String id="";
                    String name="";
                    while (iterator.hasNext()){
                        CanalEntry.Column next = iterator.next();
                        if (next.getName().equals("id")){
                            id=next.getValue();
                        }
                        if(next.getName().equals("name")){
                            name = next.getValue();
                        }
                    }

                    //float xx = 10/0;
                    String set = jedis.set(id.getBytes(StandardCharsets.UTF_8),
                            name.getBytes(StandardCharsets.UTF_8));
                    System.out.println("缓存更新:"+set);


                } else if (rowChange.getEventType() == CanalEntry.EventType.INSERT) {
                    List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();

                    System.out.println("新增行before:" + beforeColumnsList);
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    System.out.println("新增行after:" + afterColumnsList);

                }
            }
        }

    }
}
